package org.qee.mq.custom;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

/**
 * @ProjectName: learning
 * @Package: org.qee.mq.custom
 * @ClassName: MqConsumer
 * @Description:
 * @Date: 2021/12/20 10:59 上午
 * @Version: 1.0
 */
public class MqConsumer<T> implements Consumer<T> {

    private Topic topic;

    private ValueSerialization valueSerialization;

    protected Class<T> paramType;


    /**
     * 当前最大拉取物理偏移量
     */
    private long currentOffset = 0;

    // 每次最多拉10个消息
    private int batchSize = 10;

    public MqConsumer(String topicName, Class<T> typeClass) {
        this(topicName, new JsonValueSerialization());
        paramType = typeClass;
    }

    public MqConsumer(String topicName, ValueSerialization valueSerialization) {
        topic = Topic.subcribe(topicName, this);
        this.valueSerialization = valueSerialization;
    }

    @Override
    public List<T> poll() {
        return poll(currentOffset);
    }

    @Override
    public List<T> poll(long offset) {
        Queue.TakeResult takeResult = null;
        try {
            takeResult = topic.poll(offset, batchSize);
        } catch (IOException e) {
            e.printStackTrace();
            return new ArrayList<>();
        }
        if (takeResult.getResult().size() == 0) {
            return new ArrayList<>();
        }
        this.currentOffset = takeResult.getLastOffset() + 1;
        List<T> result = new ArrayList<>();
        for (byte[] bs : takeResult.getResult()) {
            try {
                if (bs.length == 0) {
                    continue;
                }
                T obj = valueSerialization.deserialize(bs, paramType);
                result.add(obj);
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        return result;
    }
}
